package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;


/**
 * @projectName: gmalll2206
 * @package: com.bw.gmall.realtime.dwd.db.app
 * @className: DwdTradeOrderCancelDetail
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/2/25 20:28
 * @version: 1.0
 */

public class DwdTradeOrderCancelDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeOrderCancelDetail().start(Constant.TOPIC_DB, Constant.TOPIC_DWD_TRADE_ORDER_CANCEL, 4, 10015);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));

        readOdsDb(tableEnv, Constant.TOPIC_DWD_TRADE_ORDER_CANCEL);
        //读取TOPIC_DWD_TRADE_ORDER_DETAIL数据
        getDwdTradeOrderDetail(tableEnv, groupId);
//        Table table = tableEnv.sqlQuery("select * from dwd_trade_order_detail");
//        table.execute().print();
        //过滤order_cancel
        Table orderCancel = getOrderCancel(tableEnv);
        tableEnv.createTemporaryView("order_cancel", orderCancel);
        tableEnv.sqlQuery("select * from order_cancel").execute().print();
        //连接
        Table result = getResult(tableEnv);
//        tableEnv.createTemporaryView("re",result);
//        Table table = tableEnv.sqlQuery("select * from re");
//        table.execute().print();
        //写入kafka
//        extracted(tableEnv);
//        result.executeInsert(Constant.TOPIC_DWD_TRADE_ORDER_CANCEL);

    }

    private static void extracted(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table "+Constant.TOPIC_DWD_TRADE_ORDER_CANCEL+"\n" +
                "(\n" +
                "    id string,\n" +
                "    order_id string,\n" +
                "    user_id string,\n" +
                "    sku_id string,\n" +
                "    sku_name string,\n" +
                "    province_id string,\n" +
                "    activity_id string,\n" +
                "    activity_rule_id string,\n" +
                "    coupon_id string,\n" +
                "    order_cancel_date_id string," +
                "    operate_time string," +
                "    sku_num string,\n" +
                "    split_original_amount DECIMAL(10, 2),\n" +
                "    split_activity_amount DECIMAL(10, 2),\n" +
                "    split_coupon_amount DECIMAL(10, 2),\n" +
                "    split_total_amount DECIMAL(10, 2),\n" +
                "    ts bigint\n" +
                ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_ORDER_CANCEL));
    }

    private static Table getResult(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery(
                "select  " +
                        "od.id," +
                        "od.order_id," +
                        "od.user_id," +
                        "od.sku_id," +
                        "od.sku_name," +
                        "od.province_id," +
                        "od.activity_id," +
                        "od.activity_rule_id," +
                        "od.coupon_id," +
                        "date_format(oc.operate_time, 'yyyy-MM-dd') order_cancel_date_id," +
                        "oc.operate_time," +
                        "od.sku_num," +
                        "od.split_original_amount," +
                        "od.split_activity_amount," +
                        "od.split_coupon_amount," +
                        "od.split_total_amount," +
                        "oc.ts " +
                        "from dwd_trade_order_detail od " +
                        "join order_cancel oc " +
                        "on od.order_id=oc.id ");
    }

    private static Table getOrderCancel(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select \n" +
                "`data`['id'] id,\n" +
                "`data`['operate_time'] operate_time,\n" +
                "`ts` \n" +
                "from topic_db \n" +
                "where `database` = 'gmall' \n" +
                " and `table` = 'order_info' \n" +
                " and `type` = 'update' \n" +
                "and `old`['order_status'] = '1001' and `data`['order_status'] = '1003'");

    }

    private static void getDwdTradeOrderDetail(StreamTableEnvironment tableEnv, String groupId) {
        tableEnv.executeSql(
                "create table " + Constant.TOPIC_DWD_TRADE_ORDER_DETAIL + "(" +
                        "id string," +
                        "order_id string," +
                        "user_id string," +
                        "sku_id string," +
                        "sku_name string," +
                        "province_id string," +
                        "activity_id string," +
                        "activity_rule_id string," +
                        "coupon_id string," +
                        "date_id string," +
                        "create_time string," +
                        "sku_num string," +
                        "split_original_amount DECIMAL(10, 2)," +
                        "split_activity_amount DECIMAL(10, 2)," +
                        "split_coupon_amount DECIMAL(10, 2)," +
                        "split_total_amount DECIMAL(10, 2)," +
                        "ts bigint" +
                        ")" + SQLUtil.getKafkaSourceSQL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, groupId));
    }
}
